package com.liwy3.reactor;

import com.liwy3.common.domain.R;
import org.reactivestreams.Subscription;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * Copyright © 元气森林（北京）食品科技集团有限公司. All rights reserved.
 *
 * @Description:
 * @author: Crazy4J
 * @date: 2022年03月01日 9:17
 * @version: V1.0
 */
public class ReactorJavaDemo {

    public static void main(String[] args) {
//        Mono byId = findById();
//        byId.subscribe(System.out::println);
//        Flux<String> fewWords = Flux.just("Hello", "World");
//        List<String> list = new ArrayList<>();
//        list.add("1");list.add("2");list.add("3");list.add("4");

//        test3();

//        test4();
        test6();

    }

    public static void test6(){
        Flux<Integer> integerFlux = Flux.range(0, 10);
        integerFlux.subscribe(new MySubscriber());
    }

    public static void test5(){
        Flux<Integer> integerFlux = Flux.range(0, 10);
        integerFlux.subscribe(i -> {
            System.out.println("run");
            System.out.println(i);
//            throw new RuntimeException();
        }, error -> {
            System.out.println("error");
        }, () -> {
            System.out.println("done");
        }, p -> {
            p.request(2);
            System.out.println("ppppppppppppppp");
        });
    }

    private static class MySubscriber extends BaseSubscriber<Integer> {

        /**
         * 初次订阅时被调用
         */
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("开始啦！");
            // 记得至少请求一次，否则不会执行hookOnNext()方法
            request(1);
        }

        /**
         * 每次读取新值调用
         */
        @Override
        protected void hookOnNext(Integer value) {
            System.out.println("开始读取...");
            System.out.println(value);
            // 指出下一次读取多少个
            request(1);
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("结束啦");
        }
    }

    public static void test4(){
        Flux<String> seq1 = Flux.just("foo", "bar", "foobar");

        List<String> iterable = Arrays.asList("foo", "bar", "foobar");
        Flux<String> seq2 = Flux.fromIterable(iterable);

        Mono<String> noData = Mono.empty();

        Mono<String> data = Mono.just("foo");

        Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);

        numbersFromFiveToSeven.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) {
                System.out.println(integer);
            }
        });

        data.subscribe(i-> System.out.println(i));

    }



    public static void test3(){
        List<String> list = new ArrayList<>();
        list.add("1");list.add("2");list.add("3");list.add("4");


        list.forEach(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(s);
            }
        });
    }

    public static void test2(){

        Mono.create(sink -> {
            ListenableFuture<ResponseEntity<String>> entity = null;

            entity.addCallback(new ListenableFutureCallback<ResponseEntity<String>>() {
                @Override
                public void onFailure(Throwable ex) {
                    sink.error(ex);
                }

                @Override
                public void onSuccess(ResponseEntity<String> result) {
                    sink.success(result.getBody());
                }
            });
        });
    }
    public static void test(){
        Flux<Integer> flux = Flux.range(0, 10);
        flux.subscribe(i -> {
            System.out.println("run1: " + i);
        });
        flux.subscribe(i -> {
            System.out.println("run2: " + i);
        });

        flux.sort().subscribe(i->{
            System.out.println("run2: " + i);
        });
        ReactorJavaDemo reactorJavaDemo = new ReactorJavaDemo();

        Mono<ReactorJavaDemo> just = Mono.just(reactorJavaDemo);
    }




    public static Mono findById() {
        WebClient webClient = WebClient.create("http://localhost:8082");
        return webClient
                .post()
                .uri("/account/add2?amount=100")
                .exchange()
                .flatMap(clientResponse -> clientResponse.bodyToMono(R.class));
    }



}
